package com.afdor.rws.event.driven.support.channel.consumer;

import com.afdor.rws.event.driven.api.EventHolder;
import com.afdor.rws.event.driven.api.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.commons.ThreadUtils;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.entities.Event;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 消费者抽象实现
 *
 * @author 悭梵
 * @date Created in 2018-12-27 9:34
 */
@Slf4j
public abstract class AbstractRunnableEventConsumer<EH extends EventHolder, CC extends EventConsumerConfig> extends AbstractEventConsumer<EH, CC> implements Runnable {

    @Override
    public final void start() {
        super.start();
        int period = getConfig().getEventHandlerPeriod();
        if (period <= 0) {
            period = 200;
        }
        // scheduleAtFixedRate  在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务。注意，period是从任务开始执行算起的。开始执行任务后，定时器每隔period时长检查该任务是否完成，如果完成则再次启动任务，否则等该任务结束后才再次启动任务
        // scheduleWithFixedDelay 在initialDelay时长后第一次执行任务，以后每当任务执行完成后，等待delay时长，再次执行任务。
        // 两个方法均在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务，但scheduleAtFixedRate以开始时间计算间隔，而scheduleWithFixedDelay以结束时间计算间隔
        ThreadUtils.getScheduledThreadPool().scheduleWithFixedDelay(this, 0, period, TimeUnit.MILLISECONDS);
    }

    @Override
    public void run() {
        if (isRunning() && !Thread.currentThread().isInterrupted()) {
            /* 线程未终止，就执行事件消费。若处理事件返回为true，这继续循环处理事件； 若处理事件返回为false，则退出执行，有计划任务重新启动执行*/
            while (consume()) {
                if (isRunning() && Thread.currentThread().isInterrupted()) {
                    break;
                }
            }
        }
    }

    @Override
    public final void stop() {
        super.stop();
        ThreadUtils.getScheduledThreadPool().remove(this);
    }
    /**
     * 获取最新事件
     *
     * @return
     */
    protected EH getEvent() {
        return null;
    }

    /**
     * 批量获取最新事件
     *
     * @param batchSize
     * @return
     */
    protected List<EH> getEvents(int batchSize) {
        return null;
    }

    /**
     * 事件确认消费成功回调
     *
     * @return
     */
    protected abstract void doSuccess(EH event, Object result);

    /**
     * 事件确认消费失败回调
     *
     * @return
     */
    protected abstract void doFailure(EH event, String message);


    /**
     * 消费具体消息
     *
     * @return 返回true表示消费到事件，可能还有更多事件待消费；false表示没有消费到事件
     */
    protected boolean consume() {
        // 批量处理
        if (getConfig().getBatchSize() > 1) {
            List<EH> eventHolderList = getEvents(getConfig().getBatchSize());
            if (eventHolderList == null || eventHolderList.isEmpty()) {
                return Boolean.FALSE;
            }
            eventHolderList.forEach(eventHolder -> handler(eventHolder));
        } else {
            // 单条处理
            EH eventHolder = getEvent();
            if (eventHolder == null || eventHolder.getEvent() == null) {
                return Boolean.FALSE;
            }
            handler(eventHolder);
        }
        return Boolean.TRUE;
    }

    /**
     * 处理单个事件消息
     *
     * @param eventHolder
     */
    protected void handler(EH eventHolder) {
        if (eventHolder == null) {
            return;
        }
        Event event = eventHolder.getEvent();
        if (event == null) {
            return;
        }
        if (tagFilter(event)) {
            try {
                event.setStatus(Event.Status.EXECUTING);
                Object result = getEventHandler().handler(event);
                doSuccess(eventHolder, result);
                event.setStatus(Event.Status.COMPLETED);
            } catch (Exception e) {
                EventRetryStrategy eventRetryStrategy = getEventRetryStrategy();
                // 返回false表示到达最大重试次数，即认为消息执行失败
                if (eventRetryStrategy != null && eventRetryStrategy.allowRetry(eventHolder)) {
                    event.setStatus(Event.Status.FAILURED);
                    eventRetryStrategy.retry(eventHolder);
                    log.error("事件[" + event.getId() + "]第" + event.getRetryCount() + "次执行失败：" + event + "\n", e);
                } else {
                    event.setStatus(Event.Status.TERMINATED);
                    doFailure(eventHolder, ThreadUtils.getStackMessage(e));
                }
            }
        }
    }

}
